package com.manbu.live.user.provider.config;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson2.JSON;
import com.manbu.live.farmework.redis.starter.keys.UserProviderCacheKeyBuilder;
import com.manbu.live.user.constants.CatchAsyncDeleteCode;
import com.manbu.live.user.dto.UserCatchAsyncDeleteDTO;
import com.manbu.live.user.dto.UserDTO;
import com.manbu.live.user.dto.UserTagDTO;
import jakarta.annotation.Resource;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.List;

/**
 * @author hanWang
 * @Title:Rocketmq 消费者配置类
 * @Package
 * @Description: 实现 InitializingBean 接口 在Spring容器中bean初始化的时候 回调到 mq的consumer的初始化
 * @date 2024/1/243:07 AM
 */

@Configuration
public class RocketMQConsumerConfig implements InitializingBean {

    private static final Logger LOGGER = LoggerFactory.getLogger(RocketMQConsumerConfig.class);
    @Resource
    private RocketMQConsumerProperties consumerProperties;
    @Resource
    private RedisTemplate<String,UserDTO> redisTemplate;


    @Resource
    private UserProviderCacheKeyBuilder userProviderCacheKeyBuilder;

    // 这个注解也可以和实现InitializingBean接口 实现一个效果  在Spring容器中bean初始化的时候 回调到 mq的consumer的初始化
    // @PostConstruct
    public void initConsumer() {
        try {
            // 初始化我们的Rocketmq消费者
            DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer();
            defaultMQPushConsumer.setConsumerGroup(consumerProperties.getGroupName());
            defaultMQPushConsumer.setNamesrvAddr(consumerProperties.getNameSrv());
            defaultMQPushConsumer.setConsumeMessageBatchMaxSize(1);
            defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            //  监听的topick
            defaultMQPushConsumer.subscribe("user-updaate-catch", "*");
            defaultMQPushConsumer.setMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                   // System.out.println(msgs.get(0).getBody());
//                    String msgStr = new String(msgs.get(0).getBody());
//                    UserDTO userDTO = JSONObject.parseObject(msgStr,UserDTO.class);
//                    if(userDTO ==null ||userDTO.getUserId() ==null){
//                        LOGGER.info("延迟删除用户信息-用户信息为空参数为空 内容为{}", msgStr);
//                    }
//                    redisTemplate.delete(keyBuilder.buildUserInfoKey(userDTO.getUserId()));
                    String json = new String(msgs.get(0).getBody());
                    UserCatchAsyncDeleteDTO userCatchAsyncDeleteDTO = JSON.parseObject(json, UserCatchAsyncDeleteDTO.class);
                    if (userCatchAsyncDeleteDTO!=null){
                        Long userId = JSON.parseObject(userCatchAsyncDeleteDTO.getJson()).getLong("userId");
                        if (CatchAsyncDeleteCode.USER_INFO_DELETE.getCode()==userCatchAsyncDeleteDTO.getCode()) {
                            redisTemplate.delete(userProviderCacheKeyBuilder.buildUserInfoKey(userId));
                        }else if (CatchAsyncDeleteCode.USER_TAG_INFO_DELETE.getCode()==userCatchAsyncDeleteDTO.getCode()) {
                            redisTemplate.delete(userProviderCacheKeyBuilder.buildUserTagInfoKey(userId));
                        }


                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }

            });
            defaultMQPushConsumer.start();
        } catch (MQClientException e) {
            throw new RuntimeException(e);
        }
        LOGGER.info("mq 消费者启动成功，namesrv is {}", consumerProperties.getNameSrv());

    }

    @Override
    public void afterPropertiesSet() throws Exception {
        initConsumer();

    }
}
